package com.zhengb.flink.study.datastream.datasource;

import com.zhengb.flink.study.datastream.entity.LocationEntity;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 模拟司机点位产生数据源
 *
 * @author Bo.Zheng1
 */
public class DriverLocationSource extends RichParallelSourceFunction<LocationEntity> {

    private volatile static boolean isRunning = true;

    private final static List<String> driverList = Arrays.asList("420356", "420357", "420358", "420359");

    @Override
    public void run(SourceContext<LocationEntity> ctx) throws Exception {

        while (isRunning) {
            ThreadLocalRandom random = ThreadLocalRandom.current();
            int randomNum = random.nextInt(10000);

            String driverNo = driverList.get(randomNum % 3);

            LocationEntity entity = new LocationEntity();
            //司机id
            entity.setDriverNo(driverNo);
            //经度
            entity.setLon(random.nextDouble());
            //纬度
            entity.setLat(random.nextDouble());

            ctx.collectWithTimestamp(entity, System.currentTimeMillis());
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {

        isRunning = false;
    }
}
